#include <unordered_set>
#include "OpBEGroupBy.h"
#include "utils/logger.h";

#include "Hasher.h"
#include "utils/logger.h"

using namespace std;

OpBEGroupBy::OpBEGroupBy(const internal::BEGroupByOperation& operation) {
	by_size = row_size = 0;
	next_row = end_row = 0;
	
	source = createOp(operation.source());
	Output in = source->getOutput();

	// przetwarzanie identyfikatorow kolumn po ktorych grupuje
	int n = operation.group_by_column().size();
	for(int i=0; i<n; i++) {
		by_column.push_back( in[operation.group_by_column(i)] );
		switch(by_column.back().type) {
			case INT:
				output.push_back( Output_elem(new char[sizeof(int32_t)*TABLESIZE],INT) );
				break;
			case BOOL:
				output.push_back( Output_elem(new char[sizeof(bool)*TABLESIZE],BOOL) );
				break;
			default: assert(false);
		}
		row_size += sizeof(int32_t);
	}
	
	by_size = row_size;
	
	// przetwarzanie identyfikatorow kolumn po ktorych jest agregacja
	n = operation.aggregations().size();
	for(int i=0; i<n; i++) {
    internal::Aggregation ag = operation.aggregations(i);
		switch(ag.type()) {
      case internal::Aggregation_Type_SUM :
				aggregated_column.push_back( in[ag.aggregated_column() ]);
				switch(aggregated_column.back().type) {
	  				case INT:
						output.push_back( Output_elem(new char[sizeof(int32_t)*TABLESIZE],INT) );
						row_size += sizeof(int32_t);
						break;
					case DOUBLE:
						output.push_back( Output_elem(new char[sizeof(double)*TABLESIZE],DOUBLE) );
						row_size += sizeof(double);
						break;
					default: assert(false);
				}
				break;
      case internal::Aggregation_Type_COUNT :
				// 0 to wartownik ktory mowi ze zamiast SUM jest COUNT (!)
				aggregated_column.push_back( Output_elem(0, INT) );
				output.push_back( Output_elem(new char[sizeof(int32_t)*TABLESIZE],INT) );
				row_size += sizeof(int32_t);
				break;
		}
	}
	
  //TODO PL - tu powinno byc chyba sizeof(int32_t), bo tyle naprawde dajemy na hasha
	row_size += sizeof(size_t); // bo dokladamy jeszcze hash'a na poczatku
	
}

int OpBEGroupBy::doit() {
	typedef unordered_set<char*, IdHasher, RowEqual > RowHashSet; RowHashSet S(TABLESIZE, IdHasher() ,
							RowEqual (by_column.size()) );
	
	
	RowHasher hashFunc(by_column.size());
	
	int size;
	while( (size = source->doit()) != -1) {
		LOG("licze hashset'a") ;
		for(int i=0; i<size; i++) {
			if(next_row == end_row) allocate_next_rows();
			
			int32_t* r = (int32_t*) ( next_row + sizeof(size_t) );			// pomijamy hasha
			for(int j=0; j<by_column.size(); j++) {
				switch(by_column[j].type) {
					case INT:
						*r = *(((int32_t*) by_column[j].column) +i);
						break;
					case BOOL:
						*r = (int32_t) *(((bool*) by_column[j].column) +i);
						break;
					default: assert(false);
				}
				++r;
			}
			
			*((size_t*) next_row) = hashFunc(next_row);
			char* q;
			pair<RowHashSet::iterator, bool> p = S.insert(next_row);
			if(p.second) { // czyli jest wlozone
				memset(r, 0, row_size - by_size - sizeof(size_t) );
				q = (char*) r;
			}
			else {
				q = (*p.first) + by_size + sizeof(size_t) ;
			}
			
			// wlasciwa agragacja
			for(int j=0; j<aggregated_column.size(); j++) {
				if(aggregated_column[j].column == 0) { // specjalny przypadek oznaczajacy COUNT
					int32_t *wsk = (int32_t*) q;
					++(*wsk) ;
					q += sizeof(int32_t);
				}
				else {
					switch(aggregated_column[j].type) {
						case INT: {
							int32_t *wsk = (int32_t*) q;
							*wsk += *(((int32_t*) aggregated_column[j].column) +i);
							q += sizeof(int32_t);
							break;
						}
						case DOUBLE: {
							double *wsk = (double*) q;
							*wsk += *(((double*) aggregated_column[j].column) +i);
							q += sizeof(double);
							break;
						}
						default: assert(false);
					}
				}
			}
			
			if(p.second) next_row += row_size;
		}
	}
	return -1;
}

void OpBEGroupBy::allocate_next_rows() {
	next_row = new char[row_size * TABLESIZE];
	rows.push_back(next_row);
	end_row = next_row + row_size * TABLESIZE;
}
